package com.stable.actor.network;

import com.stable.actor.core.ActorSystem;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 网络服务 - 处理Actor系统的所有网络通信
 * 整个ActorSystem共享一个端口
 */
@Slf4j
public class NetworkService {
    private final int port;
    private final ActorSystem actorSystem;
    private ServerSocket serverSocket;
    private ExecutorService executor;
    private final AtomicBoolean running = new AtomicBoolean(false);
    private final MessageDispatcher messageDispatcher;

    // 简化配置
    private static final int SO_TIMEOUT = 30000; // Socket读取超时30秒

    public NetworkService(int port, ActorSystem actorSystem) {
        this.port = port;
        this.actorSystem = actorSystem;
        this.messageDispatcher = new MessageDispatcher(actorSystem);
    }

    /**
     * 启动网络服务
     */
    public void start() {
        try {
            serverSocket = new ServerSocket(port);
            serverSocket.setSoTimeout(1000); // 设置accept超时，便于优雅关闭
            executor = Executors.newFixedThreadPool(20);
            running.set(true);

            log.info("🌐 ActorSystem网络服务启动在端口: {}", port);

            new Thread(this::acceptConnections, "NetworkService-" + port).start();

        } catch (IOException e) {
            log.error("❌ 启动网络服务失败，端口 {}: {}", port, e.getMessage(), e);
            throw new RuntimeException("Failed to start network service on port " + port, e);
        }
    }

    /**
     * 接受客户端连接
     */
    private void acceptConnections() {
        while (running.get()) {
            final Socket clientSocket;  // 声明为final
            try {
                clientSocket = serverSocket.accept();
                final String clientInfo = getClientInfo(clientSocket);  // 声明为final

                log.debug("🔗 接收到客户端连接: {}", clientInfo);

                clientSocket.setSoTimeout(SO_TIMEOUT);

                executor.submit(() -> handleClientConnection(clientSocket, clientInfo));

            } catch (SocketTimeoutException e) {
                continue;
            } catch (IOException e) {
                if (running.get()) {
                    log.error("❌ 接受连接失败: {}", e.getMessage(), e);
                }
            }
        }
    }


    /**
     * 获取客户端信息
     */
    private String getClientInfo(Socket socket) {
        return socket.getInetAddress().getHostAddress() + ":" + socket.getPort();
    }

    /**
     * 处理客户端连接 - 简化版本，专注于连接断开检测
     */
    private void handleClientConnection(Socket clientSocket, String clientInfo) {
        try (InputStream input = clientSocket.getInputStream();
             OutputStream output = clientSocket.getOutputStream()) {

            log.debug("📨 开始处理客户端消息: {}", clientInfo);

            while (running.get() && !clientSocket.isClosed()) {
                // 读取消息长度头 (4字节)
                byte[] header = new byte[4];
                int bytesRead = readFully(input, header);
                if (bytesRead != 4) {
                    if (bytesRead == -1) {
                        log.debug("客户端连接正常关闭: {}", clientInfo);
                    } else {
                        log.warn("❌ 读取消息头失败，期望4字节，实际: {} 字节, 客户端: {}", bytesRead, clientInfo);
                    }
                    break;
                }

                int messageLength = bytesToInt(header);
                log.debug("📏 消息长度: {} 字节, 客户端: {}", messageLength, clientInfo);

                if (messageLength <= 0 || messageLength > 10 * 1024 * 1024) {
                    log.error("❌ 无效的消息长度: {}, 客户端: {}", messageLength, clientInfo);
                    break;
                }

                // 读取消息体
                byte[] messageData = new byte[messageLength];
                bytesRead = readFully(input, messageData);
                if (bytesRead != messageLength) {
                    log.error("❌ 读取消息体失败，期望: {} 字节, 实际: {} 字节, 客户端: {}",
                            messageLength, bytesRead, clientInfo);
                    break;
                }

                // 解码消息
                ProtocolMessage message = MessageCodec.decode(messageData);
                log.debug("📨 收到消息 - 发送者: {}, 目标: {}, 客户端: {}",
                        message.getSenderPath(), message.getTargetPath(), clientInfo);

                // 处理消息
                processIncomingMessage(message, output, clientInfo);
            }

        } catch (SocketTimeoutException e) {
            log.debug("⏰ 读取超时，客户端可能已断开: {}", clientInfo);
        } catch (SocketException e) {
            log.debug("🔌 客户端连接异常断开: {}", clientInfo);
        } catch (Exception e) {
            if (running.get()) {
                log.error("❌ 处理客户端连接失败: {}", clientInfo, e);
            }
        } finally {
            // 关键：无论如何都要确保连接关闭
            closeClientSocket(clientSocket, clientInfo);
        }
    }

    /**
     * 处理接收到的消息
     */
    private void processIncomingMessage(ProtocolMessage message, OutputStream output, String clientInfo) {
        try {
            // 使用消息分发器处理消息
            messageDispatcher.dispatch(message);

            // 如果需要回复，发送响应（对于同步请求）
            if (message.getSenderPath() != null &&
                    !message.getSenderPath().equals("no-sender") &&
                    message.isExpectResponse()) {
                sendAckResponse(message, output, clientInfo);
            }

        } catch (Exception e) {
            log.error("❌ 处理消息失败: {}, 客户端: {}", message.getMessageId(), clientInfo, e);
            // 发送错误响应
            sendErrorResponse(message, e.getMessage(), output, clientInfo);
        }
    }

    /**
     * 发送确认响应
     */
    private void sendAckResponse(ProtocolMessage originalMessage, OutputStream output, String clientInfo) {
        try {
            ServiceResponse ackResponse = new ServiceResponse("ack", true, "Message processed");

            ProtocolMessage ackMessage = new ProtocolMessage();
            ackMessage.setSenderPath(originalMessage.getTargetPath());
            ackMessage.setTargetPath(originalMessage.getSenderPath());
            ackMessage.setPayload(ackResponse);
            ackMessage.setMessageId("ack-" + originalMessage.getMessageId());
            ackMessage.setTimestamp(System.currentTimeMillis());

            byte[] ackData = MessageCodec.encode(ackMessage);
            byte[] ackHeader = intToBytes(ackData.length);

            output.write(ackHeader);
            output.write(ackData);
            output.flush();

            log.debug("✅ 确认响应发送完成, 客户端: {}", clientInfo);

        } catch (IOException e) {
            log.error("❌ 发送确认响应失败，客户端可能已断开: {}", clientInfo, e);
            throw new RuntimeException("客户端连接已断开", e);
        }
    }

    /**
     * 发送错误响应
     */
    private void sendErrorResponse(ProtocolMessage originalMessage, String error, OutputStream output, String clientInfo) {
        try {
            ServiceResponse errorResponse = new ServiceResponse("error", false, error);

            ProtocolMessage errorMessage = new ProtocolMessage();
            errorMessage.setSenderPath(originalMessage.getTargetPath());
            errorMessage.setTargetPath(originalMessage.getSenderPath());
            errorMessage.setPayload(errorResponse);
            errorMessage.setMessageId("error-" + System.currentTimeMillis());
            errorMessage.setTimestamp(System.currentTimeMillis());

            byte[] errorData = MessageCodec.encode(errorMessage);
            byte[] errorHeader = intToBytes(errorData.length);

            output.write(errorHeader);
            output.write(errorData);
            output.flush();

            log.debug("⚠️ 错误响应发送完成, 客户端: {}", clientInfo);

        } catch (Exception e) {
            log.error("❌ 发送错误响应失败", e);
        }
    }

    /**
     * 发送消息到远程Actor
     */
    public void sendMessage(String host, int port, String actorPath, Object message, Object sender) {
        if (!running.get()) {
            throw new IllegalStateException("Network service is not running");
        }

        try (Socket socket = new Socket(host, port);
             OutputStream output = socket.getOutputStream();
             InputStream input = socket.getInputStream()) {

            // 创建协议消息
            ProtocolMessage protocolMessage = new ProtocolMessage();
            protocolMessage.setSenderPath(sender != null ? sender.toString() : "no-sender");
            protocolMessage.setTargetPath(actorPath);
            protocolMessage.setPayload(message);
            protocolMessage.setMessageId("msg-" + System.currentTimeMillis());
            protocolMessage.setTimestamp(System.currentTimeMillis());

            // 编码消息
            byte[] messageData = MessageCodec.encode(protocolMessage);
            byte[] header = intToBytes(messageData.length);

            // 发送消息
            output.write(header);
            output.write(messageData);
            output.flush();

            log.debug("📤 发送消息到: {}@{}:{}, 长度: {} 字节",
                    actorPath, host, port, messageData.length);

            // 等待响应（如果需要）
            if (sender != null && !"no-sender".equals(sender.toString())) {
                waitForResponse(input, protocolMessage.getMessageId());
            }

        } catch (Exception e) {
            log.error("❌ 发送消息失败: {}@{}:{}", actorPath, host, port, e);
            throw new RuntimeException("Failed to send message to " + host + ":" + port, e);
        }
    }

    /**
     * 等待响应（同步调用）
     */
    private void waitForResponse(InputStream input, String messageId) {
        try {
            // 读取响应头
            byte[] header = new byte[4];
            int bytesRead = readFully(input, header);
            if (bytesRead != 4) {
                log.warn("❌ 读取响应头失败");
                return;
            }

            int responseLength = bytesToInt(header);
            if (responseLength <= 0) {
                log.warn("❌ 无效的响应长度: {}", responseLength);
                return;
            }

            // 读取响应体
            byte[] responseData = new byte[responseLength];
            bytesRead = readFully(input, responseData);
            if (bytesRead != responseLength) {
                log.warn("❌ 读取响应体失败");
                return;
            }

            // 解码响应
            ProtocolMessage response = MessageCodec.decode(responseData);
            log.debug("📥 收到响应: {} -> {}", response.getSenderPath(), response.getTargetPath());

            // 将响应路由回发送者
            if (response.getTargetPath() != null) {
                actorSystem.sendMessage(response.getTargetPath(), response.getPayload(), null);
            }

        } catch (Exception e) {
            log.error("❌ 等待响应失败: {}", messageId, e);
        }
    }

    /**
     * 完全读取数据
     */
    private int readFully(InputStream input, byte[] buffer) throws IOException {
        int totalRead = 0;
        while (totalRead < buffer.length) {
            int read = input.read(buffer, totalRead, buffer.length - totalRead);
            if (read == -1) {
                if (totalRead == 0) {
                    return -1; // 连接关闭
                } else {
                    throw new EOFException("连接中断，已读取 " + totalRead + " 字节，期望 " + buffer.length);
                }
            }
            totalRead += read;
        }
        return totalRead;
    }

    /**
     * 安全关闭客户端Socket
     */
    private void closeClientSocket(Socket clientSocket, String clientInfo) {
        if (clientSocket != null && !clientSocket.isClosed()) {
            try {
                clientSocket.close();
                log.debug("🔌 连接已关闭: {}", clientInfo);
            } catch (IOException e) {
                log.debug("关闭连接时发生异常: {}", clientInfo, e);
            }
        }
    }

    /**
     * 工具方法：int转byte数组
     */
    private byte[] intToBytes(int value) {
        return new byte[] {
                (byte) (value >>> 24),
                (byte) (value >>> 16),
                (byte) (value >>> 8),
                (byte) value
        };
    }

    /**
     * 工具方法：byte数组转int
     */
    private int bytesToInt(byte[] bytes) {
        return ((bytes[0] & 0xFF) << 24) |
                ((bytes[1] & 0xFF) << 16) |
                ((bytes[2] & 0xFF) << 8) |
                (bytes[3] & 0xFF);
    }

    /**
     * 停止网络服务
     */
    public void shutdown() {
        running.set(false);
        try {
            if (serverSocket != null) {
                serverSocket.close();
            }
            if (executor != null) {
                executor.shutdown();
                if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                }
            }
            log.info("🛑 网络服务停止，端口: {}", port);
        } catch (Exception e) {
            log.error("❌ 停止网络服务失败", e);
        }
    }

    /**
     * 检查服务是否运行
     */
    public boolean isRunning() {
        return running.get();
    }

    /**
     * 获取服务端口
     */
    public int getPort() {
        return port;
    }
}
